"""
Case Type   : 全链路跟踪
Case Name   : 主机展示full sql的全链路信息
Create At   : 2025-09-30
Owner       : sungang14
Description :
    1、开启enable_stmt_track参数gs_guc reload -N all -I all -c "enable_stmt_track=on"
    2、开启全量sql跟踪gs_guc reload -N all -I all -c "track_stmt_stat_level = 'L1,OFF'"
    3、建表并写入数据和执行DDL、DML操作
    4、在主机表statement_history中查看trace字段select trace from statement_history where query like 'CREATE TABLE users%';
    5、解析trace字段select pg_catalog.statement_trace_decode(trace);select statement_trace_decode(trace);
    6、在主机表statement_history中查看trace字段INSERT INTO
    7、解析trace字段select pg_catalog.statement_trace_decode(trace);select statement_trace_decode(trace);
    8、在主机表statement_history中查看trace字段select
    9、解析trace字段select pg_catalog.statement_trace_decode(trace);select statement_trace_decode(trace);
    10、在主机表statement_history中查看trace字段alter
    11、解析trace字段select pg_catalog.statement_trace_decode(trace);select statement_trace_decode(trace);
    12、在主机表statement_history中查看trace字段delete
    13、解析trace字段select pg_catalog.statement_trace_decode(trace);select statement_trace_decode(trace);
    14、在主机表statement_history中查看trace字段update
    15、解析trace字段select pg_catalog.statement_trace_decode(trace);select statement_trace_decode(trace);
    16、在主机表statement_history中查看trace字段truncate
    17、解析trace字段select pg_catalog.statement_trace_decode(trace);select statement_trace_decode(trace);
    18、在主机表statement_history中查看trace字段drop
    19、解析trace字段select pg_catalog.statement_trace_decode(trace);select statement_trace_decode(trace);
Expect      :
    1、开启成功
    2、开启成功
    3、操作执行成功
    4、查看trace字段内容不为空
    5、解析trace全链路信息成功
    6、查看trace字段内容不为空
    7、解析trace全链路信息成功
    8、查看trace字段内容不为空
    9、解析trace全链路信息成功
    10、查看trace字段内容不为空
    11、解析trace全链路信息成功
    12、查看trace字段内容不为空
    13、解析trace全链路信息成功
    14、查看trace字段内容不为空
    15、解析trace全链路信息成功
    16、查看trace字段内容不为空
    17、解析trace全链路信息成功
    18、查看trace字段内容不为空
    19、解析trace全链路信息成功
History     :
"""

import os
import time
import unittest

from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger

logger = Logger()
Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class SlowSQLTestCase(unittest.TestCase):
    def setUp(self):
        logger.info(f'-----{os.path.basename(__file__)} start-----')
        self.constant = Constant()
        self.comsh = CommonSH('PrimaryDbUser')
        status = self.comsh.get_db_cluster_status('status')
        self.assertTrue(status)
        self.show_enable_stmt = 'show enable_stmt_track;'
        self.set_track_level = "track_stmt_stat_level='L1,OFF'"
        self.set_duration_statement = 'log_min_duration_statement=0'
        self.reset_track_level = "track_stmt_stat_level='OFF,L0'"
        self.reset_duration_statement = 'log_min_duration_statement=30min'
        self.show_sql_count = 'show instr_unique_sql_count;'
        self.set_sql_count = 'instr_unique_sql_count=1500'
        self.table_name = 'users_trace_Case0001'
        self.table_new_name = 'users_new_trace_Case0001'
        self.prepare_data = f"drop table if exists {self.table_name};" \
                            f"drop table if exists {self.table_new_name};" \
                            f"CREATE TABLE {self.table_name} (id SERIAL PRIMARY KEY, username VARCHAR(50) NOT NULL, email VARCHAR(100), birth_date DATE, balance NUMERIC(10,2) DEFAULT 0.00);" \
                            f"INSERT INTO {self.table_name} values (generate_series(1, 20),'name1', 'exa.com','2010-09-12', 4.6);" \
                            f"select * from {self.table_name};" \
                            f"alter table {self.table_name} RENAME TO {self.table_new_name};" \
                            f"delete from {self.table_new_name} where id=10;" \
                            f"update {self.table_new_name} set username='hahahha' where id=11;" \
                            f"truncate table {self.table_new_name};" \
                            f"drop table {self.table_new_name};"
        self.show_track_level = 'show track_stmt_stat_level;'
        self.show_duration_statement = 'show log_min_duration_statement;'
        self.prepare_data2 = f"delete from statement_history where query like '%{self.table_name}%';delete from statement_history where query like '%{self.table_new_name}%';"
        text = '--清理statement_history表内容--'
        logger.info(text)
        data_cmd = self.comsh.execut_db_sql(self.prepare_data2, dbname='postgres')
        logger.info(data_cmd)
        self.assertIn('DELETE', data_cmd, '执行失败' + text)
        show_res = self.comsh.execut_db_sql(self.show_sql_count)
        logger.info(show_res)
        self.assertEquals('100', show_res.splitlines()[-2].strip(),
                          '执行失败' + text)
        set_cmd1 = self.comsh.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            self.set_sql_count)
        logger.info(set_cmd1)

    def test_slowsql(self):
        text = '--step1:开启enable_stmt_track参数;expect:开启成功--'
        logger.info(text)
        show_cmd = self.comsh.execut_db_sql(self.show_enable_stmt)
        logger.info(show_cmd)
        self.assertEquals('on', show_cmd.splitlines()[-2].strip(),
                          '执行失败' + text)

        text = "--step2:设置track_stmt_stat_level = 'L1,OFF';expect:设置成功--"
        logger.info(text)
        set_cmd1 = self.comsh.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            self.set_track_level)
        logger.info(set_cmd1)
        show_res = self.comsh.execut_db_sql(self.show_track_level)
        logger.info(show_res)
        self.assertEquals('L1,OFF', show_res.splitlines()[-2].strip(),
                          '执行失败' + text)

        set_cmd2 = self.comsh.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            self.set_duration_statement)
        logger.info(set_cmd2)
        show_res = self.comsh.execut_db_sql(self.show_duration_statement)
        logger.info(show_res)
        self.assertEquals('0', show_res.splitlines()[-2].strip(),
                          '执行失败' + text)

        text = '--step3:建表并写入数据和执行DDL、DML操作;expect:成功--'
        logger.info(text)
        data_cmd = self.comsh.execut_db_sql(self.prepare_data, dbname='postgres')
        logger.info(data_cmd)
        self.assertIn('INSERT 0 20', data_cmd, '执行失败' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, data_cmd, '执行失败' + text)
        self.assertIn(self.constant.ALTER_TABLE_MSG, data_cmd, '执行失败' + text)
        self.assertIn(self.constant.TRUNCATE_SUCCESS_MSG, data_cmd, '执行失败' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, data_cmd, '执行失败' + text)
        self.assertIn(self.constant.UPDATE_SUCCESS_MSG, data_cmd, '执行失败' + text)
        self.assertIn(self.constant.DELETE_SUCCESS_MSG, data_cmd, '执行失败' + text)

        text = '--step4:在主机表statement_history中查看trace字段;expect:查看trace字段内容不为空--'
        logger.info(text)
        time.sleep(3)
        sql_re = self.comsh.execut_db_sql(
            f"select trace from statement_history where query like 'CREATE TABLE {self.table_name}%';",
            dbname='postgres')
        logger.info(sql_re)
        trace_msg = sql_re.splitlines()[-2].strip()
        logger.info(trace_msg)
        self.assertIsNotNone(trace_msg, '执行失败' + text)

        text = '--step5:解析trace字段;expect:解析trace全链路信息成功--'
        logger.info(text)
        sql_re1 = self.comsh.execut_db_sql(f"select pg_catalog.statement_trace_decode('{trace_msg}');",
                                           dbname='postgres')
        logger.info(sql_re1)
        self.assertIn(f"Utility query", sql_re1, '执行失败' + text)
        self.assertIn(f"ProcessUtility", sql_re1, '执行失败' + text)
        sql_re2 = self.comsh.execut_db_sql(f"select statement_trace_decode('{trace_msg}');", dbname='postgres')
        logger.info(sql_re2)
        self.assertIn(f"Utility query", sql_re2, '执行失败' + text)
        self.assertIn(f"ProcessUtility", sql_re2, '执行失败' + text)
        self.assertEqual(sql_re1, sql_re2, "执行失败" + text)

        text = '--step6:在主机表statement_history中查看trace字段;expect:查看trace字段内容不为空--'
        logger.info(text)
        sql_re = self.comsh.execut_db_sql(
            f"select trace from statement_history where query like 'INSERT INTO {self.table_name}%';",
            dbname='postgres')
        logger.info(sql_re)
        trace_msg = sql_re.splitlines()[-2].strip()
        logger.info(trace_msg)
        self.assertIsNotNone(trace_msg, '执行失败' + text)

        text = '--step7:解析trace字段;expect:解析trace全链路信息成功--'
        logger.info(text)
        sql_re1 = self.comsh.execut_db_sql(f"select pg_catalog.statement_trace_decode('{trace_msg}');",
                                           dbname='postgres')
        logger.info(sql_re1)
        self.assertIn(f"Insert query", sql_re1, '执行失败' + text)
        sql_re2 = self.comsh.execut_db_sql(f"select statement_trace_decode('{trace_msg}');", dbname='postgres')
        logger.info(sql_re2)
        self.assertIn(f"Insert query", sql_re2, '执行失败' + text)
        self.assertEqual(sql_re1, sql_re2, "执行失败" + text)

        text = '--step8:在主机表statement_history中查看trace字段;expect:查看trace字段内容不为空--'
        logger.info(text)
        sql_re = self.comsh.execut_db_sql(
            f"select trace from statement_history where query='select * from {self.table_name};';",
            dbname='postgres')
        logger.info(sql_re)
        trace_msg = sql_re.splitlines()[-2].strip()
        logger.info(trace_msg)
        self.assertIsNotNone(trace_msg, '执行失败' + text)

        text = '--step9:解析trace字段;expect:解析trace全链路信息成功--'
        logger.info(text)
        sql_re1 = self.comsh.execut_db_sql(f"select pg_catalog.statement_trace_decode('{trace_msg}');",
                                           dbname='postgres')
        logger.info(sql_re1)
        self.assertIn(f"Select query", sql_re1, '执行失败' + text)
        sql_re2 = self.comsh.execut_db_sql(f"select statement_trace_decode('{trace_msg}');", dbname='postgres')
        logger.info(sql_re2)
        self.assertIn(f"Select query", sql_re2, '执行失败' + text)
        self.assertEqual(sql_re1, sql_re2, "执行失败" + text)

        text = '--step10:在主机表statement_history中查看trace字段;expect:查看trace字段内容不为空--'
        logger.info(text)
        sql_re = self.comsh.execut_db_sql(
            f"select trace from statement_history where query='alter table {self.table_name} RENAME TO {self.table_new_name};';",
            dbname='postgres')
        logger.info(sql_re)
        trace_msg = sql_re.splitlines()[-2].strip()
        logger.info(trace_msg)
        self.assertIsNotNone(trace_msg, '执行失败' + text)

        text = '--step11:解析trace字段;expect:解析trace全链路信息成功--'
        logger.info(text)
        sql_re1 = self.comsh.execut_db_sql(f"select pg_catalog.statement_trace_decode('{trace_msg}');",
                                           dbname='postgres')
        logger.info(sql_re1)
        self.assertIn(f"Utility query", sql_re1, '执行失败' + text)
        self.assertIn(f"ProcessUtility", sql_re1, '执行失败' + text)
        sql_re2 = self.comsh.execut_db_sql(f"select statement_trace_decode('{trace_msg}');", dbname='postgres')
        logger.info(sql_re2)
        self.assertIn(f"Utility query", sql_re2, '执行失败' + text)
        self.assertIn(f"ProcessUtility", sql_re2, '执行失败' + text)
        self.assertEqual(sql_re1, sql_re2, "执行失败" + text)

        text = '--step12:在主机表statement_history中查看trace字段;expect:查看trace字段内容不为空--'
        logger.info(text)
        sql_re = self.comsh.execut_db_sql(
            f"select trace from statement_history where query like 'delete from users_new_trace_Case000%';",
            dbname='postgres')
        logger.info(sql_re)
        trace_msg = sql_re.splitlines()[-2].strip()
        logger.info(trace_msg)
        self.assertIsNotNone(trace_msg, '执行失败' + text)

        text = '--step13:解析trace字段;expect:解析trace全链路信息成功--'
        logger.info(text)
        sql_re1 = self.comsh.execut_db_sql(f"select pg_catalog.statement_trace_decode('{trace_msg}');",
                                           dbname='postgres')
        logger.info(sql_re1)
        self.assertIn(f"Delete query", sql_re1, '执行失败' + text)
        sql_re2 = self.comsh.execut_db_sql(f"select statement_trace_decode('{trace_msg}');", dbname='postgres')
        logger.info(sql_re2)
        self.assertIn(f"Delete query", sql_re2, '执行失败' + text)
        self.assertEqual(sql_re1, sql_re2, "执行失败" + text)

        text = '--step14:在主机表statement_history中查看trace字段;expect:查看trace字段内容不为空--'
        logger.info(text)
        sql_re = self.comsh.execut_db_sql(
            f"select trace from statement_history where query like 'update {self.table_new_name} set username%';",
            dbname='postgres')
        logger.info(sql_re)
        trace_msg = sql_re.splitlines()[-2].strip()
        logger.info(trace_msg)
        self.assertIsNotNone(trace_msg, '执行失败' + text)

        text = '--step15:解析trace字段;expect:解析trace全链路信息成功--'
        logger.info(text)
        sql_re1 = self.comsh.execut_db_sql(f"select pg_catalog.statement_trace_decode('{trace_msg}');",
                                           dbname='postgres')
        logger.info(sql_re1)
        self.assertIn(f"Update query", sql_re1, '执行失败' + text)
        sql_re2 = self.comsh.execut_db_sql(f"select statement_trace_decode('{trace_msg}');", dbname='postgres')
        logger.info(sql_re2)
        self.assertIn(f"Update query", sql_re2, '执行失败' + text)
        self.assertEqual(sql_re1, sql_re2, "执行失败" + text)

        text = '--step16:在主机表statement_history中查看trace字段;expect:查看trace字段内容不为空--'
        logger.info(text)
        sql_re = self.comsh.execut_db_sql(
            f"select trace from statement_history where query='truncate table {self.table_new_name};';",
            dbname='postgres')
        logger.info(sql_re)
        trace_msg = sql_re.splitlines()[-2].strip()
        logger.info(trace_msg)
        self.assertIsNotNone(trace_msg, '执行失败' + text)

        text = '--step17:解析trace字段;expect:解析trace全链路信息成功--'
        logger.info(text)
        sql_re1 = self.comsh.execut_db_sql(f"select pg_catalog.statement_trace_decode('{trace_msg}');",
                                           dbname='postgres')
        logger.info(sql_re1)
        self.assertIn(f"Utility query", sql_re1, '执行失败' + text)
        self.assertIn(f"ProcessUtility", sql_re1, '执行失败' + text)
        sql_re2 = self.comsh.execut_db_sql(f"select statement_trace_decode('{trace_msg}');", dbname='postgres')
        logger.info(sql_re2)
        self.assertIn(f"Utility query", sql_re2, '执行失败' + text)
        self.assertIn(f"ProcessUtility", sql_re2, '执行失败' + text)

        text = '--step18:在主机表statement_history中查看trace字段;expect:查看trace字段内容不为空--'
        logger.info(text)
        sql_re = self.comsh.execut_db_sql(
            f"select trace from statement_history where query = 'drop table {self.table_new_name};';",
            dbname='postgres')
        logger.info(sql_re)
        trace_msg = sql_re.splitlines()[-2].strip()
        logger.info(trace_msg)
        self.assertIsNotNone(trace_msg, '执行失败' + text)

        text = '--step19:解析trace字段;expect:解析trace全链路信息成功--'
        logger.info(text)
        sql_re1 = self.comsh.execut_db_sql(f"select pg_catalog.statement_trace_decode('{trace_msg}');",
                                           dbname='postgres')
        logger.info(sql_re1)
        self.assertIn(f"Utility query", sql_re1, '执行失败' + text)
        self.assertIn(f"ProcessUtility", sql_re1, '执行失败' + text)
        sql_re2 = self.comsh.execut_db_sql(f"select statement_trace_decode('{trace_msg}');", dbname='postgres')
        logger.info(sql_re2)
        self.assertIn(f"Utility query", sql_re2, '执行失败' + text)
        self.assertIn(f"ProcessUtility", sql_re2, '执行失败' + text)

    def tearDown(self):
        text = '--step20:清理环境，恢复默认值;expect:成功--'
        logger.info(text)
        sql_re = self.comsh.execut_db_sql(f"drop table if exists {self.table_new_name};", dbname='postgres')
        logger.info(sql_re)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, sql_re, '执行失败' + text)

        self.comsh.execute_gsguc('reload',
                                 self.constant.GSGUC_SUCCESS_MSG,
                                 self.reset_track_level)
        show_res = self.comsh.execut_db_sql(self.show_track_level)
        logger.info(show_res)
        self.assertEquals('OFF,L0', show_res.splitlines()[-2].strip(),
                          '执行失败' + text)
        self.comsh.execute_gsguc('reload',
                                 self.constant.GSGUC_SUCCESS_MSG,
                                 self.reset_duration_statement)
        show_res = self.comsh.execut_db_sql(self.show_duration_statement)
        logger.info(show_res)
        self.assertEquals('30min', show_res.splitlines()[-2].strip(),
                          '执行失败' + text)
        logger.info(f'-----{os.path.basename(__file__)} end-----')
